package org.zxp.thread.threadpool;

import org.junit.Test;
import java.util.Random;
import java.util.concurrent.*;

/**
 * @program: zxp-thread-test
 * @description: ThreadPoolExecutor测试
 * @author: X-Pacific zhang
 * @create: 2019-05-27 20:42
 **/
public class ThreadPoolExecutorTest {
    /**
     * 线程池实例化 构造方法参数
     * @param args
     */
    public static void main(String[] args){
//        new ThreadPoolExecutor(10,//核心线程池大小cSize
//                30,//线程池最大容量mSize
//                3L,//当线程数大于xSize时，多余的空闲线程在终止之前等待新任务的最大时间
//                TimeUnit.MILLISECONDS,
//                new LinkedBlockingQueue<Runnable>(),//工作队列
//                ThreadFactory,//线程工厂
//                handler)//拒绝策略

        LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(20);
        ThreadPoolExecutor excutor =  new ThreadPoolExecutor(
                10,
                30,
                3L,
                TimeUnit.SECONDS,
                queue);
//        当前的任务数X  工作队列数nWork  核心线程池大小cSize  线程池最大容量mSize
//        X <= cSize 只启动x个线程
//        启动5个线程
//        for (int i = 0; i < 5; i++) {
//            excutor.submit(() -> {
//                System.out.println(Thread.currentThread().getName());
//            });
//        }

//        X > cSize && X < nWork + cSize  (不指定工作队列大小后面条件恒为true) 会启动 <=cSize个线程，其他的在工作队列中排队
//        启动10个线程
//        for (int i = 0; i < 19; i++) {
//            excutor.submit(() -> {
//                //作用是让线程池有足够的时间创建线程
//                try {Thread.sleep(2000);} catch (InterruptedException e) {}
//                System.out.println(Thread.currentThread().getName());
//            });
//        }
//        //9个排队
//        System.out.println(queue.size());

//        X > cSize && X  > nWork + cSize  不指定工作队列后面的事情永远不会发生
//          X - nWork <=mSize 会启动 X - nWork个线程
        //启动了45-20=25个线程
//        for (int i = 0; i < 45; i++) {
//            excutor.submit(() -> {
//                //作用是让线程池有足够的时间创建线程
//                try {Thread.sleep(2000);} catch (InterruptedException e) {}
//                System.out.println(Thread.currentThread().getName());
//            });
//        }
//        //20个排队
//        System.out.println(queue.size());

//        X > cSize && X  > nWork + cSize  不指定工作队列后面的事情永远不会发生
//          X - nWork > mSize 会启动 mSize个线程,并抛出异常，执行拒绝策略
        //启动了30个线程
//        for (int i = 0; i < 70; i++) {
//            excutor.submit(() -> {
//                //作用是让线程池有足够的时间创建线程
//                try {Thread.sleep(2000);} catch (InterruptedException e) {}
//                System.out.println(Thread.currentThread().getName()+"===="+queue.size());
//            });
//        }
    }


    /**
     * 无界队列，相当于不会执行拒绝策略，所有任务都将被执行
     */
    @Test
    public void testFixPool(){
        LinkedBlockingQueue<Runnable> maxqueue = new LinkedBlockingQueue<>();
        ThreadPoolExecutor maxexcutor =  new ThreadPoolExecutor(10,30,//线程池最大容量mSize
                0,TimeUnit.SECONDS,
                maxqueue);
//        //这种不指定队列上限的，别管有多少任务都会执行
        for (int i = 0; i < 7000; i++) {
            maxexcutor.submit(() -> {
                //作用是让线程池有足够的时间创建线程
                try {Thread.sleep(2000);} catch (InterruptedException e) {}
                System.out.println(Thread.currentThread().getName() + "====" + maxqueue.size());
            });
        }
    }

    /**
     * FutureTask带返回值的单独线程运行
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void testFutureTask() throws ExecutionException, InterruptedException {
        Callable<String> callable1 = new Callable<String>() {
            @Override
            public String call() throws Exception {
                Thread.sleep(1000L);
                return "i am Callable 1";
            }
        };
        Callable<String> callable2 = new Callable<String>() {
            @Override
            public String call() throws Exception {
                Thread.sleep(2000L);
                return "i am Callable 2";
            }
        };
        FutureTask<String> futureTask1 = new FutureTask(callable1);//FutureTask 是Futrue和Runnable的子类
        FutureTask<String> futureTask2 = new FutureTask(callable2);
        new Thread(futureTask1).start();
        new Thread(futureTask2).start();
        String result1 = futureTask1.get();//阻塞1秒，直到callable1运行完毕
        System.out.println(result1);

        String result2 = futureTask2.get();//阻塞2秒，直到callable2运行完毕
        System.out.println(result2);
    }

    /**
     * 测试拒绝策略
     */
    @Test
    public void testRejectedExecutionHandler(){
//        X <= cSize
//        X > cSize && X < nWork + cSize
//        X > cSize && X  > nWork + cSize X - nWork <=mSize
//        X > cSize && X  > nWork + cSize X - nWork > mSize
        RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();
        handler = new ThreadPoolExecutor.AbortPolicy();
        handler = new ThreadPoolExecutor.CallerRunsPolicy();
        LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(1);//阻塞队列1个长度
        ThreadPoolExecutor excutor = new ThreadPoolExecutor(1,3,0L,
                        TimeUnit.SECONDS,queue,handler);
        for (int i = 0; i < 5; i++) {
            final int j = i;
            excutor.submit(() -> {
                //作用是让线程池有足够的时间创建线程
                try {Thread.sleep(2000);} catch (InterruptedException e) {}
                System.out.println(Thread.currentThread().getName() + "====" + j);
            });
        }

        //DiscardOldestPolicy要单独测试
//        handler = new ThreadPoolExecutor.DiscardOldestPolicy();
//        LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(2);//阻塞队列1个长度
//        ThreadPoolExecutor excutor = new ThreadPoolExecutor(1,3,0L,
//                TimeUnit.SECONDS,queue,handler);
//        for (int i = 0; i < 6; i++) {//任务数6 超出2+3 1个任务，这时队列中排队2个，最早加入的（未执行）扔掉，重新执行，最后有5个任务被执行
//            final int j = i;
//            excutor.submit(() -> {
////                //作用是让线程池有足够的时间创建线程
////                try {Thread.sleep(2000);} catch (InterruptedException e) {}
//                System.out.println(Thread.currentThread().getName() + "====" + j +"====" + queue.size());
//            });
//        }
        try {Thread.sleep(2000000L);} catch (InterruptedException e) {}
    }


    /**
     * ThreadFactory就是用来生产thread的工厂
     * 默认是Executors.defaultThreadFactory()，这里面除了生成线程名称还能设置为非守护线程、已经设计优先级为普通
     * 这个例子通过ThreadFactory指定线程名称
     */
    @Test
    public void testThreadFactory(){
        ThreadFactory threadFactory = r -> {
            Thread t = new Thread(r,"线程池名字"+new Random().nextInt(1000) );
            return t;
        };
        ThreadPoolExecutor excutor = new ThreadPoolExecutor(10,30,0L,
                TimeUnit.SECONDS,new LinkedBlockingQueue<>(),threadFactory);

        for (int i = 0; i < 5; i++) {
            excutor.submit(() -> {
                System.out.println(Thread.currentThread().getName());
            });
        }
        try {Thread.sleep(2000000L);} catch (InterruptedException e) {}
    }

    /**
     * 测试各种线程池执行的调用
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void testApi() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor excutor = new ThreadPoolExecutor(10,30,0L,
                TimeUnit.SECONDS,new LinkedBlockingQueue<>());
//        execute
//        excutor.execute(() -> {
//            System.out.println(Thread.currentThread().getName());
//            int a = 1/0;
//        });

//        submit
//        Future<?> submit = excutor.submit(() -> {
////            int a = 1 / 0;
//            try {Thread.sleep(3000L);} catch (InterruptedException e) {}
//            System.out.println(Thread.currentThread().getName());
//        });
//        //上面抛异常get时才捕获，上面没有返回值，这里也可以get，未执行完成会挂起
//        System.out.println(submit.get());

//        Future<String> submit = excutor.submit(() -> {
//            try {Thread.sleep(3000L);} catch (InterruptedException e) {}
//            return Thread.currentThread().getName()+"_inner";
//        });
//        //上面抛异常get时才捕获，上面没有返回值，这里也可以get，未执行完成会挂起
//        System.out.println(submit.get());

//        invokeAll
//        List<Callable<String>> list = new ArrayList();
//        for (int i = 0; i < 5; i++) {
//            final int j = i;
//            list.add(() -> {
//                int ii = 1/0;
//                try {Thread.sleep(1000L);} catch (InterruptedException e) {}
//                String msg = Thread.currentThread().getName()+"-"+j;
//                return msg;
//            });
//        }
//        List<Future<String>> results = excutor.invokeAll(list);
//        for (int i = 0; i < results.size(); i++) {
///           System.out.println("主线程输出："+results.get(i).get());
//        }

//        invokeAny 不可靠 不研究
//        List<Callable<String>> list = new ArrayList();
//        for (int i = 0; i < 5; i++) {
//            final int j = i;
//            list.add(() -> {
//                try {Thread.sleep(1000L*j);} catch (InterruptedException e) {}
//                String msg = Thread.currentThread().getName()+"-"+j;
//                System.out.println("线程内输出："+msg);
//                return msg;
//            });
//        }
//        String results = excutor.invokeAny(list);
//        System.out.println("主线程输出："+ results);
        LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
        queue.put(() -> {System.out.println("before:"+Thread.currentThread().getName());});
        queue.put(() -> {System.out.println("before:"+Thread.currentThread().getName());});
        excutor = new ThreadPoolExecutor(10,30,0L,
                TimeUnit.SECONDS,queue);
        excutor.prestartCoreThread();
//        excutor.submit(()->{System.out.println(Thread.currentThread().getName());});
        try {Thread.sleep(2000000L);} catch (InterruptedException e) {}
    }
}
